package com.atguigu.flink.day04;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;

import java.time.Duration;

/**
 * @author Felix
 * @date 2024/8/13
 * 该案例演示了Sink输出到文件
 */
public class Flink08_Sink_File {
    public static void main(String[] args) throws Exception {
        //TODO 1.指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //开启检查点
        env.enableCheckpointing(5000L);
        //TODO 2.从数据生成器中获取数据 封装为流
        DataStreamSource<String> ds1 = env.fromSource(
                new DataGeneratorSource<String>(
                        new GeneratorFunction<Long, String>() {
                            @Override
                            public String map(Long value) throws Exception {
                                return "数据:" + value;
                            }
                        },
                        Long.MAX_VALUE,
                        RateLimiterStrategy.perSecond(100),
                        Types.STRING
                ),
                WatermarkStrategy.noWatermarks(),
                "data_source"
        );
        //TODO 3.将流中数据写到文件中
        FileSink<String> sink = FileSink
                .forRowFormat(new Path("E://output//"), new SimpleStringEncoder<String>("UTF-8"))
                .withOutputFileConfig(new OutputFileConfig("atguigu-",".log"))
                .withRollingPolicy(
                        DefaultRollingPolicy.builder()
                                 //15分钟滚动一次
                                .withRolloverInterval(Duration.ofMinutes(15))
                                 //5分钟不活跃(5min没有写数据了)滚动一次
                                .withInactivityInterval(Duration.ofMinutes(5))
                                //达到指定的大小滚动一次
                                //.withMaxPartSize(MemorySize.ofMebiBytes(1024))
                                .withMaxPartSize(new MemorySize(1024))
                                .build())
                .build();
        ds1.sinkTo(sink);

        env.execute();

    }
}
